package com.millstein.realtime.app.dwd.log;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.millstein.realtime.app.base.BaseAppV1;
import com.millstein.realtime.common.Constants;
import com.millstein.realtime.util.CommonUtil;
import com.millstein.realtime.util.FlinkSinkUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.util.HashMap;
import java.util.Map;

/**
 * @Description
 * @Author tsing
 * @Date 2024-09-29 9:02
 */
@Slf4j
public class Dwd_01_BaseLogApp extends BaseAppV1 {

    private final String PAGE = "page";
    private final String ERROR = "err";
    private final String DISPLAY = "display";
    private final String ACTION = "action";
    private final String START = "start";

    public static void main(String[] args) {
        new Dwd_01_BaseLogApp().init(3001, 3, "Dwd_01_BaseLogApp", Constants.TOPIC_LOG);
    }

    /**
     * 具体数据处理的逻辑，由子类编写
     *
     * @param env          执行环境对象
     * @param streamSource 待处理的数据流
     */
    @Override
    protected void handle(StreamExecutionEnvironment env, DataStreamSource<String> streamSource) {
        // 1.数据清洗
        SingleOutputStreamOperator<JSONObject> dataStream = doEtl(streamSource);

        // 2.修正is_new字段
        dataStream = checkAndCollectNewUser(dataStream);

        // 3.利用侧输出流将原本的流根据日志类型的不同分别输出
        Map<String, DataStream<String>> streamInfo = sideOutputStreamSource(dataStream);

        // 4.将不同类型的日志流输入到kafka的不同主题中
        writeToKafkaByStream(streamInfo);
    }

    /**
     * 数据清洗，这里主要将不是json的数据过滤掉
     * @param streamSource 待清晰的数据流
     * @return
     */
    private SingleOutputStreamOperator<JSONObject> doEtl(DataStreamSource<String> streamSource) {
        return streamSource
                .filter(new FilterFunction<String>() {
                    @Override
                    public boolean filter(String json) throws Exception {
                        try {
                            JSONObject.parseObject(json);
                        } catch (Exception e) {
                            log.error("收到的数据为{}，不是json数据", json);
                            return false;
                        }
                        return true;
                    }
                })
                .map(JSONObject::parseObject);
    }

    /**
     * <p>修改新老用户的标记</p>
     * <p> 每个设备号保存一个状态，状态中的值为该设备首次访问的日期，每次处理数据的时候需要先和这个首次访问日期进行比对，</p>
     * <p> 满足条件的话就进行修正。具体的修正的逻辑如下：</p>
     * <p>     1.状态是null</p>
     * <p>         1.1.is_new是1：认为他是新用户，不需要修正，将当天的日期存入到状态中</p>
     * <p>         1.2.is_new是0：认为他不是新用户，不需要修正，将当天的前一天日期存入到状态中</p>
     * <p>     2.状态不是null</p>
     * <p>         2.1.is_new是1</p>
     * <p>             2.1.1.日志的日期和状态中的日期相同，认为他是新用户，不需要修正</p>
     * <p>             2.1.2.日志的日期和状态中的日期不同，认为他不是新用户，需要修正</p>
     * <p>         2.2.is_new是0：认为他不是新用户</p>
     * @param dataStream 待修正的数据流
     * @return
     */
    private SingleOutputStreamOperator<JSONObject> checkAndCollectNewUser(
            SingleOutputStreamOperator<JSONObject> dataStream
    ) {
        return dataStream
                .keyBy(json -> json.getJSONObject("common").getString("mid"))
                .process(new KeyedProcessFunction<String, JSONObject, JSONObject>() {
                    ValueState<String> firstVisitDateState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        firstVisitDateState = getRuntimeContext().getState(
                                new ValueStateDescriptor<>("first_visit", String.class)
                        );
                    }

                    @Override
                    public void processElement(JSONObject json, Context ctx, Collector<JSONObject> out) throws Exception {
                        String firstVisitDate = firstVisitDateState.value();

                        String isNew = json.getJSONObject("common").getString("is_new");
                        Long ts = json.getLong("ts");
                        String mid = json.getJSONObject("common").getString("mid");
                        if (firstVisitDate == null) {
                            if ("1".equals(isNew)) {
                                // 不需要修正，更新状态为今天
                                String today = CommonUtil.formatDate(ts);
                                firstVisitDateState.update(today);
                            } else {
                                // 不需要修正，更新状态为昨天
                                String yesterday = CommonUtil.formatDate(ts - 1000 * 60 * 60 * 24);
                                firstVisitDateState.update(yesterday);
                            }
                        } else {
                            String today = CommonUtil.formatDate(ts);
                            if ("1".equals(isNew) && !firstVisitDate.equals(today)) {
                                json.getJSONObject("common").put("is_new", 0);
                            }
                        }
                        out.collect(json);
                    }
                });
    }

    /**
     * 利用侧输出流将原本的流根据日志类型的不同分别输出
     * @param dataStream 待处理的数据流
     * @return
     */
    private Map<String, DataStream<String>> sideOutputStreamSource(SingleOutputStreamOperator<JSONObject> dataStream) {
        OutputTag<String> displayTag = new OutputTag<String>(DISPLAY) {};
        OutputTag<String> actionTag = new OutputTag<String>(ACTION) {};
        OutputTag<String> pageTag = new OutputTag<String>(PAGE) {};
        OutputTag<String> errorTag = new OutputTag<String>(ERROR) {};

        SingleOutputStreamOperator<String> outputStream = dataStream.process(new ProcessFunction<JSONObject, String>() {
            @Override
            public void processElement(JSONObject json, Context ctx, Collector<String> out) throws Exception {
                JSONObject common = json.getJSONObject("common");
                JSONObject page = json.getJSONObject("page");
                Long ts = json.getLong("ts");

                // 1.每个日志都可能出现err，所以要先处理
                if (json.containsKey("err")) {
                    ctx.output(errorTag, json.toJSONString());
                    json.remove("err");
                }

                // 2.启动日志不打标签，直接输出为主流
                // 因为启动日志和其他日志的格式都是不一样的。其他日志可能一条中包含多个类型，而启动日志只包含启动这一个类型
                if (json.containsKey("start")) {
                    out.collect(json.toJSONString());
                    return;
                }

                // 3.处理其他日期
                // 3.1.处理曝光日志
                if (json.containsKey("displays")) {
                    JSONArray displays = json.getJSONArray("displays");
                    for (int i = 0; i < displays.size(); i++) {
                        JSONObject displayJson = displays.getJSONObject(i);
                        displayJson.putAll(common);
                        displayJson.putAll(page);
                        displayJson.put("ts", ts);
                        ctx.output(displayTag, displayJson.toJSONString());
                    }
                    json.remove("displays");
                }


                // 3.2.处理操作日志
                if (json.containsKey("actions")) {
                    JSONArray actions = json.getJSONArray("actions");
                    for (int i = 0; i < actions.size(); i++) {
                        JSONObject actionJson = actions.getJSONObject(i);
                        actionJson.putAll(common);
                        actionJson.putAll(page);
                        ctx.output(actionTag, actionJson.toJSONString());
                    }
                    json.remove("actions");
                }

                // 3.3.处理页面日志。除了启动日志，其他日志中都有page
                if (json.containsKey("page")) {
                    ctx.output(pageTag, json.toJSONString());
                }
            }
        });

        DataStream<String> displayStream = outputStream.getSideOutput(displayTag);
        DataStream<String> actionStream = outputStream.getSideOutput(actionTag);
        DataStream<String> pageStream = outputStream.getSideOutput(pageTag);
        DataStream<String> errorStream = outputStream.getSideOutput(errorTag);

        Map<String, DataStream<String>> map = new HashMap<>();
        map.put(START, outputStream);
        map.put(DISPLAY, displayStream);
        map.put(ACTION, actionStream);
        map.put(PAGE, pageStream);
        map.put(ERROR, errorStream);

        return map;
    }

    /**
     * 将不同类型的日志流输入到kafka的不同主题中
     * @param streamInfo 待处理的数据流
     */
    private void writeToKafkaByStream(Map<String, DataStream<String>> streamInfo) {
        streamInfo.get(START).addSink(FlinkSinkUtil.getKafkaSink(Constants.TOPIC_DWD_TRAFFIC_START));
        streamInfo.get(DISPLAY).addSink(FlinkSinkUtil.getKafkaSink(Constants.TOPIC_DWD_TRAFFIC_DISPLAY));
        streamInfo.get(ACTION).addSink(FlinkSinkUtil.getKafkaSink(Constants.TOPIC_DWD_TRAFFIC_ACTION));
        streamInfo.get(PAGE).addSink(FlinkSinkUtil.getKafkaSink(Constants.TOPIC_DWD_TRAFFIC_PAGE));
        streamInfo.get(ERROR).addSink(FlinkSinkUtil.getKafkaSink(Constants.TOPIC_DWD_TRAFFIC_ERROR));
    }

}
